
package pl.edu.agh.soa.monitoring.cep;

import pl.edu.agh.soa.monitoring.CallbackUpdateListener;
import pl.edu.agh.soa.monitoring.cep.event.TestEvent;
import pl.edu.agh.soa.monitoring.router.IDynamicRouter;
import pl.edu.agh.soa.monitoring.router.entry.DynamicRouterEntry;
import pl.edu.agh.soa.monitoring.router.impl.DynamicRouter;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.EPAdministrator;
import com.espertech.esper.client.EPRuntime;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;


public class CEPProvider {

    public static interface ICEPProvider {

        public void init();

        public void registerStatement(String statement, DynamicRouterEntry entry);

        public void sendEvent(Object event);
        
        public IDynamicRouter getDynamicRouter();
    }

    public static ICEPProvider getCEPProvider() {
        String className = System.getProperty("esper.benchmark.provider", EsperCEPProvider.class.getName());
        try {
            Class clazz = Class.forName(className);
            return (ICEPProvider) clazz.newInstance();
        } catch (Throwable t) {
            t.printStackTrace();
            throw new RuntimeException(t);
        }
    }

    public static class EsperCEPProvider implements ICEPProvider {

        private EPAdministrator epAdministrator;

        private EPRuntime epRuntime;
        
        private IDynamicRouter dynamicRouter;
        
        EventInfoSubscriber eventInfoSubscriver;
        
        public EsperCEPProvider() {
        }

        public void init() {
            Configuration configuration;
            try {
                Class configurationHAClass = Class.forName("com.espertech.esperha.client.ConfigurationHA");
                configuration = (Configuration) configurationHAClass.newInstance();
                System.out.println("=== EsperHA is available, using ConfigurationHA ===");
            } catch (ClassNotFoundException e) {
                configuration = new Configuration();
            } catch (Throwable t) {
                System.err.println("Could not properly determine if EsperHA is available, default to Esper");
                t.printStackTrace();
                configuration = new Configuration();
            }
            configuration.addEventTypeAlias("EventInfo", EventInfo.class);
            configuration.addEventTypeAlias("TestEvent", TestEvent.class);
            EPServiceProvider epService = EPServiceProviderManager.getProvider("monitoring", configuration);
            epAdministrator = epService.getEPAdministrator();
            epRuntime = epService.getEPRuntime();
            
            dynamicRouter = new DynamicRouter(epAdministrator,epRuntime);
            
            eventInfoSubscriver = new EventInfoSubscriber(dynamicRouter);
            EPStatement stmt = epAdministrator.createEPL("select * from EventInfo");
            stmt.setSubscriber(eventInfoSubscriver);
        }

        public void sendEvent(Object event) {
            epRuntime.sendEvent(event);
        }

		public void registerStatement(String statementID, DynamicRouterEntry entry) {
			CallbackUpdateListener cul = new CallbackUpdateListener();
			EPStatement stmt = epAdministrator.createEPL(statementID);
			stmt.addListener(cul);
			dynamicRouter.addAddresses(statementID, entry);
		}

		public IDynamicRouter getDynamicRouter() {
			return dynamicRouter;
		}
    }
    
}
